package eureka.flink.chap10;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author com.eureka.wh
 * @since 2019/6/22 23:51
 */
public class FlatMapTest {

    public static void main(String[] args)throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(WORDS).flatMap(new RichFlatMapFunction<String, String>() {

            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] tokens = value.toLowerCase().split(",");
                for (String token: tokens){
                    if (token.length() > 0){
                        out.collect(token);
                    }
                }
            }
        }).print();

        env.execute("FlatMapTest");
    }

    public static final String[] WORDS = new String[] {
            "To be, or not to be,--that is the question:--",
            "Whether 'tis nobler in the mind to suffer",
            "The slings and arrows of outrageous fortune",
            "And by opposing end them?--To die,--to sleep,--",
            "Be all my sins remember'd."
    };
}
